package com.xiaoyushu.user.relation.biz.consumer;

import com.alibaba.nacos.shaded.com.google.common.util.concurrent.RateLimiter;
import com.xiaoyushu.framework.common.util.JsonUtils;
import com.xiaoyushu.user.relation.biz.constant.MQConstants;
import com.xiaoyushu.user.relation.biz.constant.RedisKeyConstants;
import com.xiaoyushu.user.relation.biz.domain.dataobject.FansDO;
import com.xiaoyushu.user.relation.biz.domain.dataobject.FollowingDO;
import com.xiaoyushu.user.relation.biz.domain.mapper.FansDOMapper;
import com.xiaoyushu.user.relation.biz.domain.mapper.FollowingDOMapper;
import com.xiaoyushu.user.relation.biz.enums.FollowUnfollowTypeEnum;
import com.xiaoyushu.user.relation.biz.model.dto.CountFollowUnfollowMqDTO;
import com.xiaoyushu.user.relation.biz.model.dto.FollowUserMqDTO;
import com.xiaoyushu.user.relation.biz.model.dto.UnfollowUserMqDTO;
import com.xiaoyushu.user.relation.biz.util.DateUtils;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.scripting.support.ResourceScriptSource;
import org.springframework.stereotype.Component;
import org.springframework.transaction.support.TransactionTemplate;

import java.time.LocalDateTime;
import java.util.Collections;
import java.util.Objects;

/**
 * 关注\取消关注 MQ消费者
 */
@Component
@RocketMQMessageListener(consumerGroup = "xiaoyushu_group_"+MQConstants.TOPIC_FOLLOW_OR_UNFOLLOW,//Group组
    topic = MQConstants.TOPIC_FOLLOW_OR_UNFOLLOW ,//消费的Topic 主题
    consumeMode= ConsumeMode.ORDERLY //设置为顺序消费模式
)
@Slf4j
public class FollowUnfollowConsumer implements RocketMQListener<Message> {

    @Resource
    private TransactionTemplate transactionTemplate;
    @Resource
    private  FollowingDOMapper followingDOMapper;
    @Resource
    private FansDOMapper fansDOMapper;

    //每秒创建 5000 个令牌
    @Resource
    private RateLimiter rateLimiter;

    @Resource
    private RedisTemplate redisTemplate;
    @Autowired
    private RocketMQTemplate rocketMQTemplate;


    /**
     * @param message
     */
    @Override
    public void onMessage(Message message) {
        // 流量削峰:通过获取令牌,如果没有令牌可用,将阻塞,直到获得
        rateLimiter.acquire();

        //消息体
        String bodyJsonStr = new String(message.getBody());
        //标签
        String tags = message.getTags();
        log.info("==> FollowUnfollowConsumer 消费了信息{},tags:{}",bodyJsonStr,tags);

        //根据MQ 标签,判断操作类型
        if (Objects.equals(tags,MQConstants.TAG_FOLLOW)){//关注
            handleFollowTagMessage(bodyJsonStr);
        } else if (Objects.equals(tags,MQConstants.TAG_UNFOLLOW)) {//取关
            //TODO
            handleUnfollowTagMessage(bodyJsonStr);

        }

    }

    private void handleUnfollowTagMessage(String bodyJsonStr) {
        //将消息体 Json 字符串转为DTO对象
        UnfollowUserMqDTO unfollowUserMqDTO= JsonUtils.parseObject(bodyJsonStr,UnfollowUserMqDTO.class);
        // 判空
        if (Objects.isNull(unfollowUserMqDTO)) return;;

        Long userId=unfollowUserMqDTO.getUserId();
        Long unfollowUserId=unfollowUserMqDTO.getUnfollowUserId();
        LocalDateTime createTime=unfollowUserMqDTO.getCreateTime();

        //编程式提交事务
       Boolean isSuccess= Boolean.TRUE.equals(transactionTemplate.execute(status -> {
            try{
                //取关成功需要删除数据库两条记录
                //关注表:一条记录
               int count= followingDOMapper.deleteByUserIDAndFollowingUserId(userId,unfollowUserId);

               //粉丝表:一条记录
                if (count>0){
                    fansDOMapper.deleteByUserIdAndFansUserIdInt(unfollowUserId,userId);
                }
                return true;
            }catch (Exception ex){
                status.setRollbackOnly();//标记事务为回滚
                log.error("",ex);
            }
        return  false;
        }));
        //若数据库删除成功,更新Redis,将自己从被取关用户的ZSet 粉丝列表删除
        log.info("用户:"+userId+"取关"+unfollowUserId+"博主删除粉丝");
        if (isSuccess){
            //被取关用户的粉丝列表 Redis Key
            String fanRedisKey=RedisKeyConstants.buildUserFansKey(userId);
            // 删除指定粉丝
            Long remove = redisTemplate.opsForZSet().remove(fanRedisKey, userId);
            log.info(userId+"取关"+unfollowUserId+"移除粉丝,返回结果:"+remove);

        }
    }

    /**
     * 关注
     * @param bodyJsonStr
     */
    private void handleFollowTagMessage(String bodyJsonStr) {
        //将消息体 Json字符串转为DTO 对象
        FollowUserMqDTO followUserMqDTO = JsonUtils.parseObject(bodyJsonStr, FollowUserMqDTO.class);
        //判空
        if (Objects.isNull(followUserMqDTO)) return;

        //幂等性:通过两盒唯一索引保证
        Long userId = followUserMqDTO.getUserId();
        Long followUserId = followUserMqDTO.getFollowUserId();
        LocalDateTime createTime = followUserMqDTO.getCreateTime();
        //编程式提交事务
        boolean isSuccess = Boolean.TRUE.equals(transactionTemplate.execute(status -> {
            try {
                //关注成功需往数据库添加两条记录
                //关注表:一条记录
                int count = followingDOMapper.insert(FollowingDO.builder()
                        .userId(userId)
                        .followingUserId(followUserId)
                        .createTime(createTime)
                        .build());
                //粉丝表:一条记录
                if (count > 0) {
                    fansDOMapper.insert(FansDO.builder()
                            .userId(followUserId)
                            .fansUserId(userId)
                            .createTime(createTime)
                            .build());
                }
                return true;
            } catch (Exception e) {
                status.setRollbackOnly();//标记事务为回滚
                log.error("关注报错啦!!:", e);
            }
            return false;
        }));
        log.info("## 数据库添加记录结果:{}",isSuccess);
        //更新Redis中被关注用户的ZSet 粉丝列表
        //若数据库操作成功,更新Redis 中被关注用户的ZSet 粉丝列表
        if (isSuccess){
            //Lua 脚本
            DefaultRedisScript<Long> script = new DefaultRedisScript<>();
            script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/follow_check_and_update_fans_zset.lua")));
            script.setResultType(Long.class);

            //时间戳
            long timestamp = DateUtils.localDateTime2Timestamp(createTime);

            //构建被关注用户的粉丝列表 Redis Key
            String fansRedisKey = RedisKeyConstants.buildUserFansKey(followUserId);
            //执行脚本

            Object execute = redisTemplate.execute(script, Collections.singletonList(fansRedisKey), userId, timestamp);
            log.info("博主:{}",followUserId+",添加粉丝:{}"+userId);

            //发送 MQ 通知计数服务:统计关注数
            //构建消息体 DTO
            CountFollowUnfollowMqDTO countFollowUnfollowDTO=  CountFollowUnfollowMqDTO.builder()
                    .userId(userId)
                    .targetUserId(followUserId)
                    .type(FollowUnfollowTypeEnum.FOLLOW.getCode()) //gu关注
                    .build();
            // 发送MQ
            sendMQ(countFollowUnfollowDTO);
        }
    }

    /**
     * 发送 MQ 通知计数服务
     * @param countFollowUnfollowDTO
     */
    private void sendMQ(CountFollowUnfollowMqDTO countFollowUnfollowDTO) {
        //构建消息对象,并将 DTO 转成 Json字符串设置 到消息体
        org.springframework.messaging.Message<String> message = MessageBuilder.withPayload(JsonUtils.toJsonString(countFollowUnfollowDTO)).build();
        // 异步发送MQ 消息
        rocketMQTemplate.asyncSend(MQConstants.TOPIC_COUNT_FOLLOWING, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("===> [计数服务:关注数] MQ发送成功,sendResult:{}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("===> [计数服务:关注数] MQ发送异常:",throwable);
            }
        });
        //发送MQ通知计数服务:统计粉丝数
        rocketMQTemplate.asyncSend(MQConstants.TOPIC_COUNT_FANS, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("===> [计数服务:粉丝数] MQ 发送成功,SendResult:{}",sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("===> [计数服务:粉丝数] MQ 发送异常: ",throwable);
            }
        });

    }
}
